package com.maidada.mddaichat.service.impl;

import com.github.benmanes.caffeine.cache.Cache;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.maidada.mddaichat.common.ThrowUtils;
import com.maidada.mddaichat.constant.RedisConstant;
import com.maidada.mddaichat.dto.chat.ChatSendMsgRequest;
import com.maidada.mddaichat.entity.AiModel;
import com.maidada.mddaichat.entity.UserAiConfig;
import com.maidada.mddaichat.listener.OpenAISSEEventSourceListener;
import com.maidada.mddaichat.service.ChatService;
import com.maidada.mddaichat.service.UserAiConfigService;
import com.maidada.mddaichat.service.UserService;
import com.maidada.mddaichat.util.CommonUtils;
import com.maidada.mddaichat.util.RedisUtils;
import com.unfbx.chatgpt.OpenAiStreamClient;
import com.unfbx.chatgpt.entity.chat.ChatCompletion;
import com.unfbx.chatgpt.entity.chat.Message;
import com.unfbx.chatgpt.function.KeyRandomStrategy;
import com.unfbx.chatgpt.interceptor.OpenAILogger;
import com.unfbx.chatgpt.interceptor.OpenAiResponseInterceptor;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import okhttp3.logging.HttpLoggingInterceptor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author wulinxuan
 * @date 2025-04-16 17:26
 */
@Slf4j
@Service
public class ChatServiceImpl implements ChatService {

    @Resource
    private Cache<String, SseEmitter> sseCache;

    @Resource
    private Cache<String, OpenAiStreamClient> clientCache;

    @Resource
    private Gson gson;

    @Resource
    private UserService userService;

    @Resource
    private AiModelServiceImpl aiModelService;

    @Resource
    private UserAiConfigService userAiConfigService;

    @Override
    public SseEmitter sendMessage(ChatSendMsgRequest param) {
        // 获取sse连接
        String userId = CommonUtils.getUserIdByRequest();
        SseEmitter sseEmitter = sseCache.getIfPresent(userId);
        if (Objects.isNull(sseEmitter)) {
            // 如果没有，创建一个新的sse连接
            sseEmitter = createSse(userId);
            sseCache.put(userId, sseEmitter);
        }

        String topicId = param.getTopicId();
        String msg = param.getMsg();
        String model = param.getModel();
        checkModel(model);

        // 获取历史消息
        String messageCacheKey = RedisConstant.getKey(RedisConstant.MESSAGE_CONTEXT, userId, topicId);
        String messageContext = RedisUtils.get(messageCacheKey);
        List<Message> msgHistory = new ArrayList<>();
        if (StringUtils.isNotBlank(messageContext)) {
            List<Message> messages = gson.fromJson(messageContext, new TypeToken<List<Message>>() {});
            msgHistory.addAll(messages);
        }

        // 追加当前消息
        msgHistory.add(Message.builder().content(msg).role(Message.Role.USER).build());

        // 监听器设置
        OpenAISSEEventSourceListener listener = new OpenAISSEEventSourceListener(topicId, sseEmitter);
        listener.onDone(respMessage -> {
            msgHistory.add(Message.builder().content(respMessage).role(Message.Role.ASSISTANT).build());
            // 保存会话历史
            RedisUtils.setWithDays(messageCacheKey, gson.toJson(msgHistory), 7);
        });

        // 构建请求
        ChatCompletion completion = ChatCompletion
                .builder()
                .messages(msgHistory)
                .model(model)
                .build();

        // 获取client并发起请求
        String clientCacheKey = RedisConstant.getKey(RedisConstant.CLIENT, userId, model);
        OpenAiStreamClient client = clientCache.getIfPresent(clientCacheKey);
        if (Objects.isNull(client)) {
            client = buildClient(userId, model);
            clientCache.put(clientCacheKey, client);
        }
        client.streamChatCompletion(completion, listener);

        return sseEmitter;
    }

    /**
     * 检查模型
     *
     * @param model 模型
     */
    private void checkModel(String model) {

    }

    /**
     * 创建sse连接
     *
     * @param userId 用户id
     * @return {@link SseEmitter }
     */
    private SseEmitter createSse(String userId) {
        // 默认30秒超时,设置为0L则永不超时
        SseEmitter sseEmitter = new SseEmitter(-0L);
        // 完成后回调
        sseEmitter.onCompletion(() -> {
            log.info("[{}]结束连接...................",userId);
            sseCache.invalidate(userId);
        });
        // 超时回调
        sseEmitter.onTimeout(() -> {
            log.info("[{}]连接超时...................", userId);
            sseCache.invalidate(userId);
        });
        // 异常回调
        sseEmitter.onError(
                throwable -> {
                    try {
                        log.info("[{}]连接异常,{}", userId, throwable.toString());
                        sseEmitter.send(SseEmitter.event()
                                .id(userId)
                                .name("发生异常！")
                                .data(Message.builder().content("发生异常请重试！").build())
                                .reconnectTime(3000));
                        sseCache.invalidate(userId);
                    } catch (IOException e) {
                        log.error(e.getMessage(),e);
                    }
                }
        );
        try {
            sseEmitter.send(SseEmitter.event().reconnectTime(5000));
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
        sseCache.put(userId, sseEmitter);
        log.info("[{}]创建sse连接成功！", userId);
        return sseEmitter;
    }

    private OpenAiStreamClient buildClient(String userId, String model) {
        OkHttpClient okHttpClient = buildHttpClient();
        // 查询用户配置
        UserAiConfig userAiConfig = userAiConfigService.queryUserConfig(userId, model);
        // 查询模型信息
        AiModel aiModel = aiModelService.getByModelName(model);
        String apiHost = aiModel.getHost();
        String apiKey = userAiConfig.getSecretKey();

        return OpenAiStreamClient.builder()
                .apiKey(Collections.singletonList(apiKey))
                .keyStrategy(new KeyRandomStrategy())
                .okHttpClient(okHttpClient)
                .apiHost(apiHost)
                .build();
    }

    private OkHttpClient buildHttpClient() {
        HttpLoggingInterceptor httpLoggingInterceptor = new HttpLoggingInterceptor(new OpenAILogger());
        httpLoggingInterceptor.setLevel(HttpLoggingInterceptor.Level.HEADERS);
        return (new OkHttpClient.Builder())
                .addInterceptor(httpLoggingInterceptor)
                .addInterceptor(new OpenAiResponseInterceptor())
                .connectTimeout(10L, TimeUnit.SECONDS)
                .writeTimeout(60, TimeUnit.SECONDS)
                .readTimeout(60, TimeUnit.SECONDS).build();
    }
}
